package com.example.realtimeeb.ods

/**
 * 1.从kafka的test这个topic获取数据-----FlinkKafkaConsumer
 * 2.把获取到的json格式的数据进行格式转化-----fastjson
 * database, table, type, data(jsonArray)
 * 3.把转化好的数据保存到HBase中
 *
 *
 *
 * */

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.example.realtimeeb.model.TableObject
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}


/**
 * 1.从kafka的test这个topic获取数据-----FlinkKafkaConsumer
 *
 * 2.把获取到的json格式的数据进行格式转化-----fastjson
 * type,database, table,data(jsonArray)
 *
 * 3.把转化好的数据保存到HBase中
 *
 */
object Kafka2Hbase {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(5000)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    val kafkaConsumer: FlinkKafkaConsumer[String] = new SourceKafka().getKafkaSource("ods_realdw_mysql_to_kafka")
    kafkaConsumer.setStartFromLatest()

    // 1.从kafka的test这个topic获取数据-----FlinkKafkaConsumer
    val sourceStream = env.addSource(kafkaConsumer)
//    sourceStream.print()

    // 2.把获取到的json格式的数据进行格式转化-----fastjson
    val mapped: DataStream[List[TableObject]] = sourceStream.map(line => {
      val jSONObject: JSONObject = JSON.parseObject(line)
      val database: String = classOf[String].cast(jSONObject.get("database").toString)
      val table: String = classOf[String].cast(jSONObject.get("table").toString)
      val typeInfo: String = classOf[String].cast(jSONObject.get("type").toString)
      var dataList: List[TableObject] = Nil
      jSONObject.getJSONArray("data").forEach(x => {
        dataList = TableObject(database, table, typeInfo, x.toString) :: dataList
      })
      dataList
    })

    mapped.print()

    // 3.把转化好的数据保存到HBase中
    mapped.addSink(new SinkHBase())

    env.execute()
  }

}
